package com.hmeter.hadoop.hbase;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map.Entry;
import java.util.NavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;

/**
 * Component name: HBaseGetSampler
 * Class used for HBase Get operations that supports following functionalities:
 * 1. Get value by row key, family and qualifier (possible to specify start and end row key and to retrieve values for particular qualifier for each row in specified range. Also possible to specify start row key and next number of rows to be retrieved)
 * 2. Get all values from column family by row key and family specified (possible to specify start and end row key and to retrieve all values from column family for each row in specified range. Also possible to specify start row key and next number of rows to be retrieved)
 * 3. Get all content by row key (possible to specify start and end row key and to retrieve all values for each row in specified range. Also possible to specify start row key and next number of rows to be retrieved)
 * 4. Get all content by row key from latest timestamp (same as 3rd but it retrieves all values from the latest row timestamp)
 * 5. Get all content by row key and specified filters (possible to specify start row key, end row key and filters that, if satisfied will retrieve all row content in specified range. Also possible to specify start row key and next number of rows which will be retrieved if filters satisfied)
 * 
 * @author Bakir Jusufbegovic
 * @since 0.1
 * @version 0.1
 */
public class HBaseGetSamplerUtility {
	
	private HBaseUtility hbaseUtility = null;
	private NavigableMap<String,NavigableMap<String,NavigableMap<String,String>>> allContent = new ConcurrentSkipListMap<String,NavigableMap<String,NavigableMap<String,String>>>();
	private NavigableMap<String,NavigableMap<String,String>> columnFamilyColumnQualifierContent = new ConcurrentSkipListMap<String,NavigableMap<String,String>>();
	private NavigableMap<String,String> columnQualifierContent = new ConcurrentSkipListMap<String,String>();
		
	public HBaseGetSamplerUtility(String hbaseZooKeeperQuorum, String hbaseTable) throws IOException 
	{
		hbaseUtility = new HBaseUtility(hbaseZooKeeperQuorum, hbaseTable);
	}

	public HBaseUtility getHBaseUtility()
	{
		return hbaseUtility;
	}
	
	private long calculateMostRecentTimestamp(Result r)
	{
		long maxTimestamp = Long.MIN_VALUE;		
		for (KeyValue kv : r.list()) {
			if (maxTimestamp < kv.getTimestamp()) {
				maxTimestamp = kv.getTimestamp();
			}
		}	
		return maxTimestamp;
	}
	
	//Get value by row key, family and qualifier
	public String getColumnQualifierForRangeOfRows(String startRowKey, String endRowKey, String family, String qualifier) throws IOException
	{
		hbaseUtility.setXmlContent("<rows>\n");
			
		if (startRowKey.equals(endRowKey))
		{
			Result r = hbaseUtility.getHTable().get(new Get(Bytes.toBytes(startRowKey)));
			byte[] valueObj = r.getValue(Bytes.toBytes(family), Bytes.toBytes(qualifier));
            String value = new String(valueObj);

            columnQualifierContent.put(qualifier, value);
            columnFamilyColumnQualifierContent.put(family, columnQualifierContent);
            allContent.put(startRowKey, columnFamilyColumnQualifierContent);           
            hbaseUtility.buildXmlContent(allContent);
            allContent.clear();
		}
		else
		{
			Scan scanner = new Scan(Bytes.toBytes(startRowKey),Bytes.toBytes(endRowKey));
			ResultScanner resultScan = hbaseUtility.getHTable().getScanner(scanner);

			for (Result r = resultScan.next(); r != null; r = resultScan.next()) {
	            byte[] valueObj = r.getValue(Bytes.toBytes(family), Bytes.toBytes(qualifier));
	            String value = new String(valueObj);
	            String rowValue = new String(r.getRow());

	            columnQualifierContent.put(qualifier, value);
	            columnFamilyColumnQualifierContent.put(family, columnQualifierContent);
	            allContent.put(rowValue, columnFamilyColumnQualifierContent);	            
	            hbaseUtility.buildXmlContent(allContent);
	            allContent.clear();
	        }	
			resultScan.close();
		}
		
		hbaseUtility.setXmlContent("</rows>\n");
        return hbaseUtility.getXmlContent();
	}
	
	//Get all content by row key, and family
	public String getColumnQualifiersForRangeOfRowsByColumnFamily(String startRowKey, String endRowKey, String family) throws IOException
	{
		hbaseUtility.setXmlContent("<rows>\n");
		
		if (startRowKey.equals(endRowKey))
		{
			Result r = hbaseUtility.getHTable().get(new Get(Bytes.toBytes(startRowKey)));
			NavigableMap<byte[],byte[]> mapOfData = r.getFamilyMap(Bytes.toBytes(family));
	        
	        for (Entry<byte[], byte[]> e : mapOfData.entrySet())
	        {
	        	String key = new String(e.getKey(),"UTF-8");
				String value = new String(e.getValue(),"UTF-8");			
	            columnQualifierContent.put(key, value);
	        } 
	        
	        columnFamilyColumnQualifierContent.put(family, columnQualifierContent);
	        allContent.put(startRowKey, columnFamilyColumnQualifierContent);
            hbaseUtility.buildXmlContent(allContent);
            allContent.clear();
		}
		else
		{
			Scan scanner = new Scan(Bytes.toBytes(startRowKey),Bytes.toBytes(endRowKey));
			ResultScanner resultScan = hbaseUtility.getHTable().getScanner(scanner);
						
			for (Result r = resultScan.next(); r != null; r = resultScan.next()) {
		        NavigableMap<byte[],byte[]> mapOfData = r.getFamilyMap(Bytes.toBytes(family));
		        
		        for (Entry<byte[], byte[]> e : mapOfData.entrySet())
		        {
		        	String key = new String(e.getKey(),"UTF-8");
					String value = new String(e.getValue(),"UTF-8");
					columnQualifierContent.put(key, value);
		        }
		        
		        String rowValue = new String(r.getRow());		        
		        columnFamilyColumnQualifierContent.put(family, columnQualifierContent);
		        allContent.put(rowValue, columnFamilyColumnQualifierContent);
	            hbaseUtility.buildXmlContent(allContent);
	            allContent.clear();
		    }
			
			resultScan.close();
		}	
		
		hbaseUtility.setXmlContent("</rows>\n");
		return hbaseUtility.getXmlContent();
	}
	
	//Get all content by row key
	public String getAllColumnQualifiersForRangeOfRows(String startRowKey, String endRowKey) throws IOException
	{
		hbaseUtility.setXmlContent("<rows>\n");
		
		if (startRowKey.equals(endRowKey))
		{
			Result r = hbaseUtility.getHTable().get(new Get(Bytes.toBytes(startRowKey)));
			
			NavigableMap<byte[],NavigableMap<byte[],byte[]>> allallContent = r.getNoVersionMap();
			
			for (byte[] cF : allallContent.keySet())
			{
				String columnFamily = new String(cF,"UTF-8");
				NavigableMap<byte[],byte[]> mapOfData = r.getFamilyMap(cF);
	            
	            for (Entry<byte[], byte[]> e : mapOfData.entrySet())
	            {
	            	String key = new String(e.getKey(),"UTF-8");
					String value = new String(e.getValue(),"UTF-8");
					columnQualifierContent.put(key, value);
	            }
	            
	            columnFamilyColumnQualifierContent.put(columnFamily, columnQualifierContent);
	            columnQualifierContent = new ConcurrentSkipListMap<String,String>();
			}
			allContent.put(startRowKey, columnFamilyColumnQualifierContent);
            hbaseUtility.buildXmlContent(allContent);
            allContent.clear();	
		}
		else
		{
			Scan scanner = new Scan(Bytes.toBytes(startRowKey),Bytes.toBytes(endRowKey));
			ResultScanner resultScan = hbaseUtility.getHTable().getScanner(scanner);
					
			for (Result r = resultScan.next(); r != null; r = resultScan.next()) {
				
				NavigableMap<byte[],NavigableMap<byte[],byte[]>> allallContent = r.getNoVersionMap();
				
				for (byte[] cF : allallContent.keySet())
				{
					String columnFamily = new String(cF,"UTF-8");
					NavigableMap<byte[],byte[]> mapOfData = r.getFamilyMap(cF);
		            
		            for (Entry<byte[], byte[]> e : mapOfData.entrySet())
		            {
		            	String key = new String(e.getKey(),"UTF-8");
						String value = new String(e.getValue(),"UTF-8");
						columnQualifierContent.put(key, value);
		            }
		            
		            columnFamilyColumnQualifierContent.put(columnFamily, columnQualifierContent);
		            columnQualifierContent = new ConcurrentSkipListMap<String,String>();
				}
				
				String rowValue = new String(r.getRow());
				allContent.put(rowValue, columnFamilyColumnQualifierContent);
				hbaseUtility.buildXmlContent(allContent);
	            allContent.clear();
		    }          
			resultScan.close();
		}
		
		hbaseUtility.setXmlContent("</rows>\n");
		return hbaseUtility.getXmlContent();
	}
		
	//Get all content by row key from latest timestamp
	public String getAllColumnQualifiersForRangeOfRowsByMostRecentTimestamp(String startRowKey, String endRowKey) throws IOException
	{
		hbaseUtility.setXmlContent("<rows>\n");
		
		if (startRowKey.equals(endRowKey))
		{
			Result r = hbaseUtility.getHTable().get(new Get(Bytes.toBytes(startRowKey)));				
			NavigableMap<byte[],NavigableMap<byte[],NavigableMap<Long,byte[]>>> allVersionedallContent = r.getMap();
			long maxTs = this.calculateMostRecentTimestamp(r);
			
			for (byte[] cF : allVersionedallContent.keySet())
			{
				String columnFamily = new String(cF,"UTF-8");
				NavigableMap<byte[],NavigableMap<Long,byte[]>> qualifierMap = allVersionedallContent.get(cF);
				
				for (Entry<byte[], NavigableMap<Long, byte[]>> e : qualifierMap.entrySet()) 
				{
					NavigableMap<Long,byte[]> valueMap = e.getValue();
					byte[] value = valueMap.get(maxTs);
		            
		            if (value != null)
		            {
		            	String key = new String(e.getKey(),"UTF-8");
						String strValue = new String(value,"UTF-8");
						columnQualifierContent.put(key, strValue);
		            }       
				}
				columnFamilyColumnQualifierContent.put(columnFamily, columnQualifierContent);
				columnQualifierContent = new ConcurrentSkipListMap<String,String>();
			}
			allContent.put(startRowKey, columnFamilyColumnQualifierContent);
            hbaseUtility.buildXmlContent(allContent);
            allContent.clear();
		}
		else
		{
			Scan scanner = new Scan(Bytes.toBytes(startRowKey),Bytes.toBytes(endRowKey));
			ResultScanner resultScan = hbaseUtility.getHTable().getScanner(scanner);
					
			for (Result r = resultScan.next(); r != null; r = resultScan.next()) {
				
				NavigableMap<byte[],NavigableMap<byte[],NavigableMap<Long,byte[]>>> allVersionedallContent = r.getMap();
				long maxTs = this.calculateMostRecentTimestamp(r);
				
				for (byte[] cF : allVersionedallContent.keySet())
				{
					String columnFamily = new String(cF,"UTF-8");
					NavigableMap<byte[],NavigableMap<Long,byte[]>> qualifierMap = allVersionedallContent.get(cF);
					
					for (Entry<byte[], NavigableMap<Long, byte[]>> e : qualifierMap.entrySet()) 
					{
						NavigableMap<Long,byte[]> valueMap = e.getValue();
						byte[] value = valueMap.get(maxTs);
			            
			            if (value != null)
			            {
			            	String key = new String(e.getKey(),"UTF-8");
							String strValue = new String(value,"UTF-8");
							columnQualifierContent.put(key, strValue);
			            }       
					}
					columnFamilyColumnQualifierContent.put(columnFamily, columnQualifierContent);
					columnQualifierContent = new ConcurrentSkipListMap<String,String>();
				}
				
				String rowValue = new String(r.getRow());
				allContent.put(rowValue, columnFamilyColumnQualifierContent);
	            hbaseUtility.buildXmlContent(allContent);    
	            allContent.clear();
			}
			resultScan.close();
		}
		
		hbaseUtility.setXmlContent("</rows>\n");
		return hbaseUtility.getXmlContent();
	}
	
	//Get value by row key, family and qualifier (specified number of next rows retrieved starting with row key that is specified)
	public String getColumnQualifierForRangeOfRows_NumberOfRowsToRetrieve(String startRowKey, int numberOfRows, String family, String qualifier) throws IOException
	{
		hbaseUtility.setXmlContent("<rows>\n");
			
		Scan scanner = new Scan(Bytes.toBytes(startRowKey));
		ResultScanner resultScan = hbaseUtility.getHTable().getScanner(scanner);
		Result[] results = resultScan.next(numberOfRows);
			
		for (Result r : results)
		{
			byte[] valueObj = r.getValue(Bytes.toBytes(family), Bytes.toBytes(qualifier));
	        String value = new String(valueObj);
	        String rowValue = new String(r.getRow());

	        columnQualifierContent.put(qualifier, value);
	        columnFamilyColumnQualifierContent.put(family, columnQualifierContent);
	        allContent.put(rowValue, columnFamilyColumnQualifierContent);	            
	        hbaseUtility.buildXmlContent(allContent);
	        allContent.clear();
		}

		resultScan.close();
		
		hbaseUtility.setXmlContent("</rows>");
		return hbaseUtility.getXmlContent();
	}
	
	//Get all content by row key, and family (specified number of next rows retrieved starting with row key that is specified)
	public String getColumnQualifiersForRangeOfRowsByColumnFamily_NumberOfRowsToRetrieve(String startRowKey, int numberOfRows, String family) throws IOException
	{
		hbaseUtility.setXmlContent("<rows>\n");
		
		Scan scanner = new Scan(Bytes.toBytes(startRowKey));
		ResultScanner resultScan = hbaseUtility.getHTable().getScanner(scanner);
		Result[] results = resultScan.next(numberOfRows);
						
		for (Result r : results) 
		{
		    NavigableMap<byte[],byte[]> mapOfData = r.getFamilyMap(Bytes.toBytes(family));
		        
		    for (Entry<byte[], byte[]> e : mapOfData.entrySet())
		    {
		        	String key = new String(e.getKey(),"UTF-8");
					String value = new String(e.getValue(),"UTF-8");
					columnQualifierContent.put(key, value);
		    }
		        
		    String rowValue = new String(r.getRow());		        
		    columnFamilyColumnQualifierContent.put(family, columnQualifierContent);
		    allContent.put(rowValue, columnFamilyColumnQualifierContent);
		    hbaseUtility.buildXmlContent(allContent);
		    allContent.clear();
		}
			
		resultScan.close();	
		hbaseUtility.setXmlContent("</rows>\n");
		return hbaseUtility.getXmlContent();
	}
	
	//Get all content by row key (specified number of next rows retrieved starting with row key that is specified)
	public String getAllColumnQualifiersForRangeOfRows_NumberOfRowsToRetrieve(String startRowKey, int numberOfRows) throws IOException
	{
		hbaseUtility.setXmlContent("<rows>\n");

		Scan scanner = new Scan(Bytes.toBytes(startRowKey));
		ResultScanner resultScan = hbaseUtility.getHTable().getScanner(scanner);
		Result[] results = resultScan.next(numberOfRows);
 			
		for (Result r : results) 
		{		
			NavigableMap<byte[],NavigableMap<byte[],byte[]>> allallContent = r.getNoVersionMap();
				
			for (byte[] cF : allallContent.keySet())
			{
				String columnFamily = new String(cF,"UTF-8");
				NavigableMap<byte[],byte[]> mapOfData = r.getFamilyMap(cF);
		            
		        for (Entry<byte[], byte[]> e : mapOfData.entrySet())
		        {
		            String key = new String(e.getKey(),"UTF-8");
					String value = new String(e.getValue(),"UTF-8");
					columnQualifierContent.put(key, value);
		        }
		            
		        columnFamilyColumnQualifierContent.put(columnFamily, columnQualifierContent);
		        columnQualifierContent = new ConcurrentSkipListMap<String,String>();
			}
				
			String rowValue = new String(r.getRow());
			allContent.put(rowValue, columnFamilyColumnQualifierContent);
			hbaseUtility.buildXmlContent(allContent);
	        allContent.clear();
		}          
		resultScan.close();
		
		hbaseUtility.setXmlContent("</rows>\n");
		return hbaseUtility.getXmlContent();
	}
	
	//Get all content by row key from latest timestamp (specified number of next rows retrieved starting with row key that is specified)
	public String getAllColumnQualifiersForRangeOfRowsByMostRecentTimestamp_NumberOfRowsToRetrieve(String startRowKey, int numberOfRows) throws IOException
	{
		hbaseUtility.setXmlContent("<rows>\n");
		
		Scan scanner = new Scan(Bytes.toBytes(startRowKey));
		ResultScanner resultScan = hbaseUtility.getHTable().getScanner(scanner);
		Result[] results = resultScan.next(numberOfRows);
			
		for (Result r : results) {
				
			NavigableMap<byte[],NavigableMap<byte[],NavigableMap<Long,byte[]>>> allVersionedallContent = r.getMap();
			long maxTs = this.calculateMostRecentTimestamp(r);
				
			for (byte[] cF : allVersionedallContent.keySet())
			{
				String columnFamily = new String(cF,"UTF-8");
				NavigableMap<byte[],NavigableMap<Long,byte[]>> qualifierMap = allVersionedallContent.get(cF);
					
				for (Entry<byte[], NavigableMap<Long, byte[]>> e : qualifierMap.entrySet()) 
				{
					NavigableMap<Long,byte[]> valueMap = e.getValue();
					byte[] value = valueMap.get(maxTs);
			            
			        if (value != null)
			        {
			            String key = new String(e.getKey(),"UTF-8");
						String strValue = new String(value,"UTF-8");
						columnQualifierContent.put(key, strValue);
			        }       
				}
				columnFamilyColumnQualifierContent.put(columnFamily, columnQualifierContent);
				columnQualifierContent = new ConcurrentSkipListMap<String,String>();
			}
				
			String rowValue = new String(r.getRow());
			allContent.put(rowValue, columnFamilyColumnQualifierContent);
	        hbaseUtility.buildXmlContent(allContent);    
	        allContent.clear();
		}
		resultScan.close();
		
		hbaseUtility.setXmlContent("</rows>\n");
		return hbaseUtility.getXmlContent();
	}
	
	//Get all content by row key and specified filters
	public String getRowsContentByFilters(String startRowKey, String stopRowKey, String operator, String filters, String delimiter) throws Exception
	{
		hbaseUtility.setXmlContent("<rows>\n");
		
		Filter filter = null;
		
		ArrayList<String> filtersList = hbaseUtility.parseElementsToList(filters, delimiter);
		HBaseFilterComparator filterComparator = new HBaseFilterComparator();
		
		String columnFamilyColumnValueToken[] = null;
		String columnValueToken[] = null;
		
		Scan scanner = new Scan(Bytes.toBytes(startRowKey),Bytes.toBytes(stopRowKey));
		List<Filter> allFilters = new ArrayList<Filter>();
	
		for (int i=0; i<filtersList.size(); i++) {
			filterComparator.validateComparatorInFullColumnQualifier(filtersList.get(i));		
			columnFamilyColumnValueToken = filtersList.get(i).split(":");
			columnValueToken = columnFamilyColumnValueToken[1].split(filterComparator.getComparator());
			
			filter = new SingleColumnValueFilter(Bytes.toBytes(columnFamilyColumnValueToken[0]), Bytes.toBytes(columnValueToken[0]), filterComparator.getCompareOp(), Bytes.toBytes(columnValueToken[1]));
			allFilters.add(filter);				
		}
		
		FilterList finalFilters = new FilterList(FilterList.Operator.valueOf(operator), allFilters);			
		scanner.setFilter(finalFilters);		
		ResultScanner resultScan = hbaseUtility.getHTable().getScanner(scanner);
	
		for (Result r = resultScan.next(); r != null; r = resultScan.next())
		{		
			NavigableMap<byte[],NavigableMap<byte[],byte[]>> allallContent = r.getNoVersionMap();
				
			for (byte[] cF : allallContent.keySet())
			{
				String columnFamily = new String(cF,"UTF-8");
				NavigableMap<byte[],byte[]> mapOfData = r.getFamilyMap(cF);
		            
		        for (Entry<byte[], byte[]> e : mapOfData.entrySet())
		        {
		            String key = new String(e.getKey(),"UTF-8");
					String value = new String(e.getValue(),"UTF-8");
					columnQualifierContent.put(key, value);
		        }
		            
		        columnFamilyColumnQualifierContent.put(columnFamily, columnQualifierContent);
		        columnQualifierContent = new ConcurrentSkipListMap<String,String>();
			}
				
			String rowValue = new String(r.getRow());
			allContent.put(rowValue, columnFamilyColumnQualifierContent);
			hbaseUtility.buildXmlContent(allContent);
	        allContent.clear();
		}          
		resultScan.close();
		
		hbaseUtility.setXmlContent("</rows>\n");
		return hbaseUtility.getXmlContent();
	}
	
	//Get all content by row key and specified filters (specified number of next rows for filtering starting with row key that is specified)
	public String getRowsContentByFilters_NumberOfRowsToRetrieve(String startRowKey, int numberOfRows, String operator, String filters, String delimiter) throws Exception
	{
		hbaseUtility.setXmlContent("<rows>\n");
		
		ArrayList<String> filtersList = hbaseUtility.parseElementsToList(filters, delimiter);
		HBaseFilterComparator filterComparator = new HBaseFilterComparator();
		
		String columnFamilyColumnValueToken[] = null;
		String columnValueToken[] = null;
		
		Scan scanner = new Scan(Bytes.toBytes(startRowKey));
		List<Filter> allFilters = new ArrayList<Filter>();
			
		for (int i=0; i<filtersList.size(); i++) {
			filterComparator.validateComparatorInFullColumnQualifier(filtersList.get(i));		
			columnFamilyColumnValueToken = filtersList.get(i).split(":");
			columnValueToken = columnFamilyColumnValueToken[1].split(filterComparator.getComparator());
			
			Filter filter = new SingleColumnValueFilter(Bytes.toBytes(columnFamilyColumnValueToken[0]), Bytes.toBytes(columnValueToken[0]), filterComparator.getCompareOp(), Bytes.toBytes(columnValueToken[1]));
			allFilters.add(filter);				
		}
		
		FilterList filterList = new FilterList(FilterList.Operator.valueOf(operator), allFilters);
		
		scanner.setFilter(filterList);
		
		ResultScanner resultScan = hbaseUtility.getHTable().getScanner(scanner);
		Result[] results = resultScan.next(numberOfRows);
	
		for (Result r : results) 
		{		
			NavigableMap<byte[],NavigableMap<byte[],byte[]>> allallContent = r.getNoVersionMap();
				
			for (byte[] cF : allallContent.keySet())
			{
				String columnFamily = new String(cF,"UTF-8");
				NavigableMap<byte[],byte[]> mapOfData = r.getFamilyMap(cF);
		            
		        for (Entry<byte[], byte[]> e : mapOfData.entrySet())
		        {
		            String key = new String(e.getKey(),"UTF-8");
					String value = new String(e.getValue(),"UTF-8");
					columnQualifierContent.put(key, value);
		        }
		            
		        columnFamilyColumnQualifierContent.put(columnFamily, columnQualifierContent);
		        columnQualifierContent = new ConcurrentSkipListMap<String,String>();
			}
				
			String rowValue = new String(r.getRow());
			allContent.put(rowValue, columnFamilyColumnQualifierContent);
			hbaseUtility.buildXmlContent(allContent);
	        allContent.clear();
		}          
		resultScan.close();
		
		hbaseUtility.setXmlContent("</rows>\n");
		return hbaseUtility.getXmlContent();
	}
}
